package com.neusoft.bd.myflink.mytable.step1;

import org.apache.flink.table.api.*;

import static org.apache.flink.table.api.Expressions.$;

public class QueryWithTableAPI {

    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnvironment = TableEnvironment.create(settings);
<<<<<<< HEAD
        tableEnvironment.executeSql( " CREATE TABLE KafkaTable (\n" +
                "  `sid` STRING,\n" +
                "  `call_in` STRING,\n" +
                "  `call_out` STRING,\n" +
                "  `call_type` STRING,\n" +
                "  `call_time` BIGINT,\n" +
                "  `duration` BIGINT \n" +

                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'test',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'csv'\n" +
                ") ");

        tableEnvironment.executeSql(  " select  sid, sum(duration)  sd   from   KafkaTable   " +
                "   where    duration>=10 and  call_type ='success' " +
                "   group by sid   ").print();

    }



=======
        tableEnvironment.executeSql("  CREATE TABLE KafkaTable (\n" +
                "  `sid` string,\n" +
                "  `call_in` STRING,\n" +
                "  `call_out` STRING,\n" +
                "  `call_type` STRING,\n" +
                "  `duraition` bigint,\n" +


                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'user_behavior',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'csv'\n" +
                ")  ");
//       通过SQL
        tableEnvironment.executeSql("  select sid , sum( duration ) as td from   KafkaTable where call_type ='sucess' and duration >=10  " +
                "group by sid  ").print();

    }

>>>>>>> a794acd2a61a93808fcd4fb5f501fbdd21a190e5
    public static void main1(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnvironment = TableEnvironment.create(settings);
<<<<<<< HEAD
        tableEnvironment.createTemporaryTable("KafkaTable" ,
=======
        tableEnvironment.createTemporaryTable("KafkaTable",
>>>>>>> a794acd2a61a93808fcd4fb5f501fbdd21a190e5
                TableDescriptor.forConnector("kafka")
                        .schema(Schema.newBuilder()
                                .column("sid", DataTypes.STRING())
                                .column("call_in", DataTypes.STRING())
                                .column("call_out", DataTypes.STRING())
                                .column("call_type", DataTypes.STRING())
                                .column("call_time", DataTypes.STRING())
                                .column("duraition", DataTypes.STRING())
<<<<<<< HEAD
                                                              .build() )
                        .option("topic","test")
                        .option("properties.bootstrap.servers","localhost:9092")
                        .option("properties.group.id","testGroup")
                        .option("scan.startup.mode","latest-offset") // latest-offset
                        .option("format" ,"csv")
=======
                                .build())
                        .option("topic", "stationlog-topic")
                        .option("properties.bootstrap.servers", "localhost:9092")
                        .option("properties.group.id", "testGroup")
                        .option("scan.startup.mode", "earliest-offset") // latest-offset
                        .option("format", "csv")
>>>>>>> a794acd2a61a93808fcd4fb5f501fbdd21a190e5
                        .build());
//        读取数据
//        tableEnvironment.from("KafkaTable").execute().print();
        Table table = tableEnvironment.from("KafkaTable");
        table.filter($("call_type").isEqual("succes").and($("duraition").isGreater(10)))
                .groupBy($("sid"))
                .select($("sid"), $("duraition").sum().as("统计时长"))
                .execute().print();


    }
}
